package com.ot.flink.cdc.function.mapper;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ot.flink.cdc.dto.TableLogDto;
import com.ot.flink.cdc.function.sink.CustomSink;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 更新数据明细
 * @author: admin
 * @date: 2024/5/8
 * @since JDK1.8 version: 1.0.0
 */
public class TableLogMapper extends RichMapFunction<String, TableLogDto> {
    private static Logger LOGGER = LoggerFactory.getLogger(TableLogMapper.class);

    @Override
    public TableLogDto map(String value) throws Exception {
        JSONObject allValue = JSON.parseObject(value);
        TableLogDto dto = new TableLogDto();
        dto.setTs(allValue.getLong("ts_ms"));
        JSONObject rowObject;
        if (null != (rowObject = allValue.getJSONObject("before"))) {
            dto.setBefore(rowObject);
        }
        if (null != (rowObject = allValue.getJSONObject("after"))) {
            dto.setBefore(rowObject);
        }
        dto.setCurrent(StringUtils.isNoneBlank(dto.getTable()) ? dto.getAfter() : dto.getBefore());
        final String op = allValue.getString("op");
        dto.setOp(op);
        JSONObject jsonSource = allValue.getJSONObject("source");
        dto.setDatabase(jsonSource.getString("db"));
        dto.setTable(jsonSource.getString("table"));
        return dto;
    }

}
